package com.example.myapplicationndk.rxjava;

import android.annotation.SuppressLint;
import android.util.Log;

import androidx.annotation.NonNull;

import com.example.myapplicationndk.utils.GenerateString;

import org.jetbrains.annotations.NotNull;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.Random;
import java.util.concurrent.TimeUnit;

import io.reactivex.Flowable;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.Scheduler;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.functions.Function3;
import io.reactivex.schedulers.Schedulers;

public class RxUtils {

    private static String tag = RxUtils.class.getSimpleName();

    public static void Rxjava() {

        Observer<String> observer = new Observer<String>() {

            @Override
            public void onSubscribe(Disposable d) {
                Log.d(tag, "onSubscribe " + d.isDisposed());
            }

            @Override
            public void onNext(String s) {
                Log.d(tag, "onNext " + s.toString());
            }

            @Override
            public void onError(Throwable e) {
                Log.d(tag, "onError " + e.toString());
            }

            @Override
            public void onComplete() {
                Log.d(tag, "onComplete ");
            }
        };

        Observable observable = Observable.create(new ObservableOnSubscribe<String>() {

            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("Hello");
                e.onNext("Hi");
                e.onNext("Aloha");
                e.onComplete();
            }
        }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread());
        Thread t;
        observable.subscribe(observer);
    }

    public static void interval() {

        Observable.interval(1, TimeUnit.SECONDS)
                .map(new Function<Long, Long>() {

                    @Override
                    public Long apply(@NotNull Long aLong) throws Exception {
                        return aLong * 10;
                    }
                })
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Log.d("taggg", aLong + GenerateString.getRandomString(new Random().nextInt(10)));
                    }
                });

    }

    public static void map() {

        Observable.just("123123").map(new Function<String, String>() {
            @Override
            public String apply(@NotNull String s) throws Exception {
                return s + GenerateString.getRandomString(new Random().nextInt(10));
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.d(tag, s);
            }
        });
    }

    public static void flatmap() {

        Observable.just("123", "456", "789").flatMap(new Function<String, ObservableSource<Integer>>() {
            @Override
            public ObservableSource<Integer> apply(String s) throws Exception {

                String[] strings = s.split("");
                //产生新的Observable
                return Observable.fromArray(strings).map(new Function<String, Integer>() {
                    @Override
                    public Integer apply(String strings) throws Exception {
                        return Integer.valueOf(strings) * 100;
                    }
                });
            }
        }).repeat(100)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(tag, GenerateString.getRandomString(10));
                    }
                });
    }

    public static void CountDowner() {
//        Flowable.intervalRange(0, 60, 0, 1, TimeUnit.SECONDS)
//                .observeOn(AndroidSchedulers.mainThread())
//                .doOnNext(new Consumer<Long>() {
//                    @Override
//                    public void accept(Long aLong) throws Exception {
//                        Log.d(tag, "time over left " + (60 - aLong));
//                    }
//                })
//                .doOnComplete(new Action() {
//                    @Override
//                    public void run() throws Exception {
//                        //倒计时完毕事件处理
//                        Log.d(tag, "time over");
//                    }
//                })
//                .subscribe();

        Observable.intervalRange(0, 60, 0, 1, TimeUnit.SECONDS)
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Log.d(tag, "time over left " + (60 - aLong));
                    }
                });
    }

    public static void switchThread() {
        CompositeDisposable comDisposable = new CompositeDisposable();

        Observable<String> observable = Observable
                .create(new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(@NonNull ObservableEmitter<String> emitter) throws
                            Exception {
                        emitter.onNext("hello");
                    }
                })
                .map(new Function<String, String>() {
                    @Override
                    public String apply(String s) throws Exception {
                        return s;
                    }
                })
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread());

        observable.subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                comDisposable.add(d);
            }

            @Override
            public void onNext(String s) {
                Log.i(tag, s);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    /**
     * String 转为 Integer 然后转换发射源 发射 String 在然后 将 String
     */
    public static void SwitchThread2() {
        Observable
                .create(new ObservableOnSubscribe<Object>() {
                    @Override
                    public void subscribe(@NotNull ObservableEmitter<Object> emitter) throws Exception {
                        Log.e(tag, "===subscribe: " + Thread.currentThread().getName());
                        emitter.onNext("1");
                    }
                })
                .map(new Function<Object, Object>() {
                    @Override
                    public Object apply(@NotNull Object o) throws Exception {
                        Log.e(tag, "===apply: " + Thread.currentThread().getName() + " " + o.toString());
                        return o;
                    }
                })
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<Object>() {
                    @Override
                    public void onSubscribe(@NotNull Disposable d) {

                    }

                    @Override
                    public void onNext(@NotNull Object o) {
                        Log.e(tag, "===onNext: " + Thread.currentThread().getName() + " " + o.toString());
                    }

                    @Override
                    public void onError(@NotNull Throwable e) {
                        Log.e(tag, "===onError: " + Thread.currentThread().getName() + " " + e.toString());
                    }

                    @Override
                    public void onComplete() {
                        Log.e(tag, "===onComplete: " + Thread.currentThread().getName());
                    }
                });
    }
}
